package geotrellis.chatta

import com.typesafe.config.ConfigFactory
import geotrellis.raster.Tile
import geotrellis.spark.SpatialKey
import geotrellis.spark.etl.Etl
import geotrellis.spark._
import geotrellis.spark.util.SparkUtils
import geotrellis.vector.ProjectedExtent
import org.apache.spark.SparkConf

object ChattaIngest {
  def main(args: Array[String]): Unit = {
    val args = new Array[String](6)
    args(0)="--input"
    args(1)="conf/input.json"
    args(2)="--output"
    args(3)="conf/output.json"
    args(4)="--backend-profiles"
    args(5)="conf/backend-profiles.json"
    goETL(args)
  }

  def goETL(args:Array[String]):Unit = {
    implicit val sc = SparkUtils.createSparkContext(
      "GeoTrellis ETL SinglebandIngest",
      new SparkConf(true)
        .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
        .set("spark.kryo.registrator", "geotrellis.spark.io.kryo.KryoRegistrator")
        .setMaster(ConfigFactory.load().getString("spark.master "))
    )
    try {
      //持久保存使用（ProjectedExtent,SpatialKey），单波段(Tile)
      //Etl.ingest[ProjectedExtent, SpatialKey, Tile]
      //持久保存使用（ProjectedExtent,SpatialKey），多波段(MultibandTile)
      //Etl.ingest[ProjectedExtent, SpatialKey, MultibandTile]
      //临时保存在内存中使用（TemporalProjectedExtent,SpaceTimeKey），单波段(Tile)
      //Etl.ingest[TemporalProjectedExtent, SpaceTimeKey, Tile]
      //Etl.ingest[TemporalProjectedExtent, SpaceTimeKey, MultibandTile]
      Etl.ingest[ProjectedExtent, SpatialKey, Tile](args)
    } finally {
      sc.stop()
    }
  }

}
